/*
 * Copyright (c) 2022-2023 cetcclout Co., Ltd All rights reserved.
 * Sensor Adapter is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *          http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */
package com.cetccloud.cimp.adapter.server.netty;

import cn.hutool.core.io.BufferUtil;
import com.cetccloud.cimp.adapter.cache.ClientDeviceCache;
import com.cetccloud.cimp.adapter.cache.MqttCenter;
import com.cetccloud.cimp.adapter.datasource.redis.RedisService;
import com.cetccloud.cimp.adapter.handler.DataContextValue;
import com.cetccloud.cimp.adapter.protocol.chain.DefaultDataChain;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * @author
 * @apiNote Netty数据接收处理服务
 * @date 2022/12/28 14:48
 */
@Component
@ChannelHandler.Sharable
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    private static final InternalLogger logger = InternalLoggerFactory.getInstance(NettyServerHandler.class);
    private final DefaultDataChain dataChain;

    public static NettyServerHandler nettyServerHandler;
    @Autowired
    private RedisService redisService;

    public NettyServerHandler(DefaultDataChain dataChain) {
        this.dataChain = dataChain;
    }

    /**
     * 初始化配置
     */
    @PostConstruct
    public void init() {
        logger.info("初始化相关配置信息！");
        nettyServerHandler = this;
        nettyServerHandler.redisService  = this.redisService;
    }

    /**
     * 处理接收到数据
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        try {
            ByteBuf in = (ByteBuf) msg;
            byte[] bytes = ByteBufUtil.getBytes(in);

            //信道数据转发到处理链路中，进行链路处理
            dataChain.getAuthHandler().fromDtu(BufferUtil.create(bytes), ctx);

        } catch (Exception e) {
            e.printStackTrace();
            logger.error("数据处理出现异常：" + e);
        } finally {
            // 处理完毕后抛弃收到的数据
            ReferenceCountUtil.release(msg);
        }
    }

    /**
     * 建立连接
     */
    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        logger.info("注册链接：设备上线==> " + ctx.channel());
        super.channelRegistered(ctx);
    }

    /**
     * 建立连接时，返回消息
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        logger.info("通道活跃：信道正常==> " + ctx.channel());
        super.channelActive(ctx);
    }

    /**
     * 切断已有链接
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        logger.info("通道关闭：信道关闭==> " + ctx.channel());
        CloseMqttClient(ctx);
        super.channelInactive(ctx);
    }

    /**
     * 关闭连接
     */
    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        logger.info("断开链接：设备下线==> " + ctx.channel());
        CloseMqttClient(ctx);
        super.channelUnregistered(ctx);
    }

    /**
     * 信道异常，关闭链接
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.error("链接异常：信道关闭==> " + ctx.channel() + cause);
        CloseMqttClient(ctx);
        super.exceptionCaught(ctx,cause);
    }

    /**
     * 关闭发布数据链接
     */
    public void CloseMqttClient(ChannelHandlerContext ctx) throws MqttException {
        // 当出现异常就关闭连接
        if (ClientDeviceCache.containDtuId(ctx.channel())) {
            try {
                String dtuId = ClientDeviceCache.getDtuId(ctx.channel());
                ctx.channel().close();
                ClientDeviceCache.removeMap(dtuId);
                //关闭mqtt链接——设备下线
                MqttClient sampleClient= MqttCenter.findMqttClient(dtuId);
                logger.info("查询MQTT链接：：{}",sampleClient);
                if(sampleClient.isConnected()){
                    sampleClient.disconnect();
                    sampleClient.close();
                }
                MqttCenter.removeMqttClient(dtuId);
                logger.info(sampleClient.getClientId()+"::Mqtt连接关闭");
                //清除缓存集合中的设备
                nettyServerHandler.redisService.setRemove(DataContextValue.DTU_STATUS_KEY,dtuId);
            } catch (MqttException e) {
                e.printStackTrace();
                logger.error("链接异常：信道关闭==>MqttException::"+e.getMessage());
            }
        }
    }

}
